package com.lighting.shortlink.project.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.lighting.shortlink.project.common.convention.exception.ServiceException;
import com.lighting.shortlink.project.dao.entity.*;
import com.lighting.shortlink.project.dao.mapper.*;
import com.lighting.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.lighting.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

import static com.lighting.shortlink.project.common.constant.RedisKeyConstant.LOCK_GID_UPDATE_KEY;
import static com.lighting.shortlink.project.common.constant.ShortLinkConstant.AMAP_REMOTE_URL;

/**
 * 短链接监控状态保存消息队列消费者
 * @author : lighting
 */

@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(
    topic = "${rocketmq.producer.topic}",
    consumerGroup = "${rocketmq.producer.group}"
)
// 实现了 RocketMQListener<Map<String, String>> 接口，表示它能够处理从RocketMQ消息队列中接收到的 Map类型的消息
public class ShortLinkStatsSaveConsumer implements RocketMQListener<Map<String,String>> {

    private final ShortLinkMapper shortLinkMapper;
    private final ShortLinkGotoMapper shortLinkGotoMapper;
    private final RedissonClient redissonClient;
    private final LinkAccessStatsMapper linkAccessStatsMapper;
    private final LinkLocaleStatsMapper linkLocaleStatsMapper;
    private final LinkOsStatsMapper linkOsStatsMapper;
    private final LinkBrowserStatsMapper linkBrowserStatsMapper;
    private final LinkAccessLogsMapper linkAccessLogsMapper;
    private final LinkDeviceStatsMapper linkDeviceStatsMapper;
    private final LinkNetworkStatsMapper linkNetworkStatsMapper;
    private final LinkStatsTodayMapper linkStatsTodayMapper;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;


    @Value("${short-link.stats.locale.amap-key}")
    private String statsLocaleAmapKey;//高德地图的位置API key

    @Override
//    在 onMessage 方法中，首先检查消息是否已经被处理过
//    如果已经处理过，则直接返回。如果没有处理过，检查是否
//    已经完成处理流程，如果是，则返回。如果不是，则抛出异常，表示需要消息队列重试。
    public void onMessage(Map<String, String> producerMap) {
        String keys = producerMap.get("keys");
        if(messageQueueIdempotentHandler.isMessageProcessed(keys))//已经在消费了（进行中）
        {
            if(messageQueueIdempotentHandler.isAccomplish(keys))//已经消费完成了
                return;
            throw new ServiceException("消息未完成处理流程，需要消息队列重试");//消息被消费了，但是目前还没有处理完成
        }
        try {
//            此处开始消费消息
            String fullShortUrl = producerMap.get("fullShortUrl");
            if(StrUtil.isNotBlank(fullShortUrl)){
                String gid = producerMap.get("gid");
                ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
                actualSaveShortLinkStats(fullShortUrl,gid,statsRecord);
            }
        } catch (Throwable ex) {
            log.error("记录短链接监控消费异常",ex);
            throw ex;
        }
//        顺利走完上面流程，标记此消息消费处理完成
        messageQueueIdempotentHandler.setAccomplish(keys);
    }

    /**
     *  记录 长连接的统计信息
     * @param fullShortUrl 长链接
     * @param gid 分组 id
     * @param statsRecord 统计信息记录
     */
    private void actualSaveShortLinkStats(String fullShortUrl, String gid, ShortLinkStatsRecordDTO statsRecord) {
//        如果 fullShortUrl 为 null，则使用 statsRecord 对象中的 getFullShortUrl() 方法获取的值作为 fullShortUrl 的值
        fullShortUrl = Optional.ofNullable(fullShortUrl).orElse(statsRecord.getFullShortUrl());
//        锁 short-link:lock:update-gid:fullShortUrl
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_GID_UPDATE_KEY, fullShortUrl));
        RLock rLock = readWriteLock.readLock();//获取读锁
        rLock.lock();//加读锁,读读不互斥，读写互斥
        try {
//             gid 为空的情况，就通过 fullShortUrl 先把 gid 查出来
            if (StrUtil.isBlank(gid)) {
                Wrappers.lambdaQuery(ShortLinkGotoDO.class).eq(ShortLinkGotoDO::getFullShortUrl,fullShortUrl);
                LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class)
                        .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
                ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);
                gid = shortLinkGotoDO.getGid();
            }
            int hour = DateUtil.hour(new Date(), true);//获取当前小时
            Week week = DateUtil.dayOfWeekEnum(new Date());//获取当前时间是周几 例：week = THURSDAY
            int weekValue = week.getIso8601Value();//获取当前周的 SOo 8601 值 例：4
            LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                    .pv(1)// 访问量+1
//                    分情况：第一次访问 1，重复访问 0
                    .uv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .uip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .hour(hour)
                    .weekday(weekValue)
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .date(new Date())
                    .build();
            linkAccessStatsMapper.shortLinkStats(linkAccessStatsDO);// 记录基础访问监控数据

//           创建一个 hasMmap 用于存储 高德API key 和 ip地址 参数，发起get请求获取 ip 地址的当前位置信息
            Map<String, Object> localeParamMap = new HashMap<>();
            localeParamMap.put("key", statsLocaleAmapKey);
            localeParamMap.put("ip", statsRecord.getRemoteAddr());
//            调用高德地图的API 获取 位置信息
            String localeResultStr = HttpUtil.get(AMAP_REMOTE_URL, localeParamMap);
            JSONObject localeResultObj = JSON.parseObject(localeResultStr);
            String infoCode = localeResultObj.getString("infocode");//获取信息码
            String actualProvince = "未知";//实际省份
            String actualCity = "未知";//实际城市

//                        正常返回的情况
            if (StrUtil.isNotBlank(infoCode) && StrUtil.equals(infoCode, "10000")) {
                String province = localeResultObj.getString("province");//获取返回的省份（有可能没查到）
                boolean unknownFlag = StrUtil.equals(province, "[]");//判断是否查到结果了
                LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder()
                        .province(actualProvince = unknownFlag ? actualProvince : province)//高德查到了就用查到的，没查到就用默认的 “未知”
                        .city(actualCity = unknownFlag ? actualCity : localeResultObj.getString("city"))
                        .adcode(unknownFlag ? "未知" : localeResultObj.getString("adcode"))//地址码
                        .cnt(1)
                        .fullShortUrl(fullShortUrl)
                        .country("中国")
                        .gid(gid)
                        .date(new Date())
                        .build();
                linkLocaleStatsMapper.shortLinkLocaleState(linkLocaleStatsDO);//记录地区访问监控数据
            }
//            记录操作系统访问监控数据
            LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                    .os(statsRecord.getOs())
                    .cnt(1)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkOsStatsMapper.shortLinkOsState(linkOsStatsDO);
//            记录浏览器访问监控数据
            LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                    .browser(statsRecord.getBrowser())
                    .cnt(1)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkBrowserStatsMapper.shortLinkBrowserState(linkBrowserStatsDO);
//            记录访问设备监控数据
            LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                    .device(statsRecord.getDevice())
                    .cnt(1)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);
//            记录访问网络监控数据
            LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                    .network(statsRecord.getNetwork())
                    .cnt(1)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);
//          记录汇总访问日志
            LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                    .user(statsRecord.getUv())
                    .ip(statsRecord.getRemoteAddr())
                    .browser(statsRecord.getBrowser())
                    .os(statsRecord.getOs())
                    .network(statsRecord.getNetwork())
                    .device(statsRecord.getDevice())
                    .locale(StrUtil.join("-", "中国", actualProvince, actualCity))// 用 “-” 拼接 “中国” 省份 城市
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .build();
            linkAccessLogsMapper.insert(linkAccessLogsDO);

//            短链接访问统计自增
            shortLinkMapper.incrementStats(gid, fullShortUrl, 1, statsRecord.getUvFirstFlag() ? 1 : 0, statsRecord.getUipFirstFlag() ? 1 : 0);

//            记录今日统计监控数据
            LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder()
                    .todayPv(1)
                    .todayUv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .todayUip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkStatsTodayMapper.shortLinkTodayState(linkStatsTodayDO);
        } catch (Throwable ex) {
            log.error("短链接访问量统计异常", ex);
        } finally {
            rLock.unlock();
        }
    }
}
